package org.jetlinks.community.messaging.rabbitmq.impl;

import org.jetlinks.community.messaging.rabbitmq.AdminRabbitProperties;
import org.jetlinks.community.messaging.rabbitmq.RabbitMQDataSource;
import org.jetlinks.community.messaging.rabbitmq.command.CreateConsumer;
import org.jetlinks.community.messaging.rabbitmq.command.CreateNativeSender;
import org.jetlinks.community.messaging.rabbitmq.command.CreateProducer;
import org.jetlinks.community.datasource.AbstractDataSource;
import org.jetlinks.community.datasource.Command;
import org.jetlinks.community.datasource.DataSourceType;
import org.jetlinks.community.messaging.rabbitmq.command.admin.AdminCommand;
import org.springframework.util.StringUtils;
import org.springframework.web.reactive.function.client.WebClient;

class DefaultRabbitMQDataSource extends AbstractDataSource<AdminRabbitProperties> implements RabbitMQDataSource {

    private WebClient adminClient;

    public DefaultRabbitMQDataSource(String id,
                                     AdminRabbitProperties config) {
        super(id, config);
        registerHandler(CreateConsumer.class, new CreateConsumerHandler());
        registerHandler(CreateProducer.class, new CreateProducerHandler());
        registerHandler(CreateNativeSender.class, new CreateNativeSenderHandler());
        initAdminClient();
    }

    @Override
    public DataSourceType getType() {
        return RabbitMQDataSource.type;
    }

    @Override
    protected <R> R handleNoHandlerCommand(Command<R> command) {
        if (command instanceof AdminCommand) {
            if (adminClient == null) {
                throw new UnsupportedOperationException("Unsupported RabbitMQ admin operations");
            }
            return ((AdminCommand<R>) command).doRequest(adminClient);
        }
        return super.handleNoHandlerCommand(command);
    }

    @Override
    protected void handleSetConfig(AdminRabbitProperties oldConfig, AdminRabbitProperties newConfig) {

        initAdminClient();
    }

    private void initAdminClient() {
        if (StringUtils.isEmpty(getConfig().getAdminUrl())) {
            return;
        }
        adminClient = WebClient
            .builder()
            .baseUrl(getConfig().getAdminUrl())
            .defaultHeaders(headers -> headers.setBasicAuth(getConfig().getUsername(), getConfig().getPassword()))
            .build();
    }
}
